package com.yonyou.pmclouds.async.injvm;


import com.yonyou.pmclouds.async.AsyncInfoInnerQuery;
import com.yonyou.pmclouds.async.AsyncInvocationInfoMaintain;
import com.yonyou.pmclouds.async.DefaultAsyncInvocationFactory;
import com.yonyou.pmclouds.basecom.async.AsyncInvocation;
import com.yonyou.pmclouds.basecom.async.AsyncInvocationInfo;
import com.yonyou.pmclouds.basecom.async.InvokingTracer;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.util.Date;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 异步调用执行器,根据AsyncInvocationInfo生成AsyncInvocation，进入线程池执行
 *
 * @see AsyncInvocation
 * @see AsyncInvocationInfo
 */
@Slf4j
public class AsyncInvocationQueueExecutor implements ApplicationContextAware {
    private static ConcurrentHashMap<String, AsyncInvocation.AsyncInvocationFactory> factoryCache = new ConcurrentHashMap<>();
    private static AsyncInvocation.AsyncInvocationFactory DEFAULT_FACTORY = new DefaultAsyncInvocationFactory();
    private static AtomicInteger threadNo = new AtomicInteger(0);
    @Autowired
    private InJvmAsynInvocationConfiguration.InJvmAsyncProperties properties;

    @Autowired
    private AsyncInvocationInfoMaintain maintain;
    private AsyncInfoInnerQuery query;
    private ApplicationContext applicationContext;

    private volatile ThreadPoolExecutor threadPoolExecutor = null;


    public void submit(AsyncInvocationInfo infoToSubmit) {
        if (infoToSubmit == null
                || StringUtils.isEmpty(infoToSubmit.getInvocationId())
                || StringUtils.isEmpty(infoToSubmit.getExecutionClass())) {
            return;
        }
        infoToSubmit.setState(AsyncInvocation.InvocationState.STATE_PREPARED);
        infoToSubmit.setSubmitTime(new Date());
        // 根据info生成具体的invocation
        AsyncInvocation invocation = null;
        String factoryClz = infoToSubmit.getExecutionClass() + "Factory";
        AsyncInvocation.AsyncInvocationFactory factory = factoryCache.get(factoryClz);

        if (factory == null) {
            // 这里考虑结合spring,优先从spring中获取工厂
            try {
                factory = (AsyncInvocation.AsyncInvocationFactory) applicationContext.getBean(factoryClz);
                factoryCache.putIfAbsent(factoryClz, factory);
            } catch (Exception e) {
                // 吃掉spring创建bean时的异常
            }
        }
        if (factory == null) {
            try {
                Class<?> invocationFactoryClz = Class.forName(factoryClz);
                factory = (AsyncInvocation.AsyncInvocationFactory) invocationFactoryClz.newInstance();
                factoryCache.putIfAbsent(factoryClz, factory);
            } catch (ClassNotFoundException | InstantiationException | IllegalAccessException e) {
                factory = DEFAULT_FACTORY;
                factoryCache.putIfAbsent(factoryClz, factory);
            }
        }
        invocation = factory.createInvocationFromInfo(infoToSubmit);
        getExecutorService().submit(new Invoker(invocation, maintain, query));
    }

    /**
     * 获取线程池对象
     */
    private ThreadPoolExecutor getExecutorService() {
        if (threadPoolExecutor == null) {
            synchronized (this) {
                if (threadPoolExecutor == null) {
                    threadPoolExecutor = new ThreadPoolExecutor(properties.getCorePoolSize(),
                            properties.getMaximumPoolSize(),
                            properties.getMaximumPoolSize(),
                            TimeUnit.SECONDS,
                            new LinkedBlockingDeque<Runnable>(properties.getMaxWaitInvocation()),
                            new ThreadFactory() {
                                @Override
                                public Thread newThread(Runnable r) {
                                    Thread t = new Thread(r);
                                    t.setName("async-thread-" + threadNo.incrementAndGet());
                                    t.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
                                        @Override
                                        public void uncaughtException(Thread t, Throwable e) {
                                            log.error("async uncaught exception occur", e);
                                        }
                                    });
                                    return t;
                                }
                            },
                            new RejectedExecutionHandler() {
                                @Override
                                public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                                    log.info("async invoke push to queue failed ,waiting schedule");
                                }
                            });
                }
            }
        }

        return threadPoolExecutor;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
    }

    @Slf4j
    private static class Invoker implements Runnable {
        private final AsyncInvocation invocation;
        private final AsyncInvocationInfoMaintain maintain;
        private final InvokingTracer tracer;
        private AsyncInfoInnerQuery query;

        Invoker(final AsyncInvocation invocation, final AsyncInvocationInfoMaintain maintain, final AsyncInfoInnerQuery query) {

            this.invocation = invocation;
            this.maintain = maintain;
            this.query = query;
            this.tracer = new InvokingTracer() {
                @Override
                public void trace(String step, AsyncInvocationInfo info) {
                    // 可以考虑丰富trace的内容
                    Invoker.this.maintain.updateInfo(info);

                }

                @Override
                public AsyncInvocationInfo fetchNewestInfo(String id) {
                    return Invoker.this.query.queryFullInfoById(id);
                }
            };
        }

        @Override
        public void run() {
            try {
                int resumeState = invocation.prepare(tracer);
                if (resumeState != AsyncInvocation.InvocationState.STATE_INVOKING) {
                    log.info("async execution does not execute, {}", invocation.getClass());
                    return;
                }
                int invokedState = invocation.invoke(tracer);
                if (invokedState != AsyncInvocation.InvocationState.STATE_SUCCESS) {
                    // 后续需要根据状态来做处理
                    log.info("async execution failed,{}", invocation.getClass());
                }
            } catch (Exception ex) {
                invocation.onException(ex, tracer);
            } finally {
                invocation.cleanResource(tracer);
            }
        }
    }
}
